// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <cantor/catalog/database_manager.h>
#include <cantor/catalog/table_manager.h>
#include <cantor/catalog/database_querier.h>

namespace cantor::catalog {
    turbo::Status TableManager::create_table(std::shared_ptr<TableInfo> info, CatalogStorage *storage,
                                             DatabaseManager *db_manager) {
        int64_t db_id = -1;
        int64_t temp_id = -1;
        {
            std::unique_lock lk(_mutex);

            auto it = _table_id_map.find(info->database_name());
            if (it != _table_id_map.end()) {
                return turbo::already_exists_error("database ", info->database_name(), "already exists with id:",
                                                   it->second);
            }
            DatabaseQuerier db_q(db_manager);

            TURBO_MOVE_OR_RAISE(db_id, db_q.database_id(info->namespace_name()));

            info->set_database_id(db_id);
            /// save to storage
            temp_id = _max_table_id + 1;
            info->set_table_id(temp_id);
            auto key = storage->make_table_key(info->table_name(), temp_id);
            auto value = info->serialize();
            TURBO_RETURN_NOT_OK(storage->save_table(key, value, temp_id));
            /// update memory
            _max_table_id = temp_id;
            _table_id_map[info->table_name()] = temp_id;
            _table_map[temp_id] = std::move(info);
        }

        return db_manager->add_table_id(db_id, temp_id);
    }

    turbo::Status TableManager::remove_table(const std::string &db_name, CatalogStorage *storage,
                                             DatabaseManager *db_manager) {
        int64_t db_id = -1;
        int64_t temp_id = -1;
        {
            std::unique_lock lk(_mutex);

            auto it = _table_id_map.find(db_name);
            if (it == _table_id_map.end()) {
                return turbo::not_found_error("database ", db_name, " not found");
            }

            temp_id = it->second;

            auto &info = _table_map[it->second];

            DatabaseQuerier ns_q(db_manager);

            TURBO_MOVE_OR_RAISE(db_id, ns_q.database_id(info->database_name()));

            /// save to storage
            auto key = storage->make_table_key(info->table_name(), info->table_id());
            TURBO_RETURN_NOT_OK(storage->remove_table(key));
            /// update memory
            _table_id_map.erase(info->table_name());
            _table_map.erase(temp_id);
        } {
            return db_manager->remove_table_id(db_id, temp_id);
        }
    }

    turbo::Status TableManager::update_table(std::shared_ptr<TableInfo> info, CatalogStorage *storage) {
        std::unique_lock lk(_mutex);

        auto it = _table_id_map.find(info->database_name());
        if (it == _table_id_map.end()) {
            return turbo::not_found_error("not found database ", info->database_name());
        }

        auto id = it->second;
        auto &old = _table_map[id];
        TURBO_ABORT_NOT_OK(info->merge_for_update(old.get()));

        auto key = storage->make_table_key(info->table_name(), id);
        auto value = info->serialize();
        TURBO_RETURN_NOT_OK(storage->update_table(key, value));

        /// update memory
        _table_map[id] = std::move(info);
        return turbo::OkStatus();
    }

    void TableManager::set_max_id(const int64_t max_id) {
        std::unique_lock lk(_mutex);
        _max_table_id = max_id;
    }
} // namespace cantor::catalog
